package net.htjs.flinkcdc.process;

import com.alibaba.fastjson.JSONObject;
import net.htjs.flinkcdc.entity.FcythJkLog;
import net.htjs.flinkcdc.service.FcythJkLogService;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

@Service
public class TableDataProcessFunction extends ProcessFunction<String, String> {

    private static final Logger logger = LoggerFactory.getLogger(TableDataProcessFunction.class);

    private FcythJkLogService fcythJkLogService;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 手动从 Spring 容器获取 Bean
        fcythJkLogService = SpringContextUtil.getBean(FcythJkLogService.class);
    }

    @Override
    public void processElement(String jsonData, Context context, Collector<String> collector) throws Exception {
        // 解析CDC数据（Debezium的JSON格式）
        JSONObject data = JSONObject.parseObject(jsonData);

        // 获取元数据中的表名
        String tableName = data.getJSONObject("source").getString("table");
        logger.info("Received data from table: {}", tableName);

        // 根据表名进行不同处理
        switch (tableName) {
            case "fcyth_jk_log":
                processFgjJkLogTable(data);
                break;
            default:
                logger.warn("Unknown table: {}", tableName);
        }

        // 可以将处理后的数据发送到下游
        collector.collect(data.toString());
    }

    // 处理产品表数据
    private void processFgjJkLogTable(JSONObject data) {
        try {
            JSONObject afterData = data.getJSONObject("after");
            // 产品表特定处理逻辑...
            logger.info("Processing 接口日志 data: {}", afterData);
            FcythJkLog item = JSONObject.parseObject(afterData.toJSONString(), FcythJkLog.class);
            fcythJkLogService.save(item);
        } catch (Exception e) {
            logger.error("保存 FcythJkLog 数据失败: ", e);
        }
    }
}
